package com.sk.rpcAssistNetty.io.netty.client;

import com.sk.rpcAssistNetty.io.netty.DefaultFuture;
import com.sk.rpcAssistNetty.io.netty.RpcRequest;
import com.sk.rpcAssistNetty.io.netty.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.jboss.netty.channel.ChannelHandler;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * description：netty客户端处理器(入站出站io处理器)
 * 【功能】
 *      ①：【wait-notify异步接收结果机制】服务调用的结果通过回调函数通知的方法返回，也就是说调用的方法处得不到返回结果，所以需要使用wait-notify机制
 *      ②：【发送消息】在发送消息的时候记录请求对象的标志值，形成键值映射，方便后面notify后获取对应的返回值
 *      ③：【回调消息】接收回调消息
 *
 * @author 孙文斌
 * @Date: 2021/3/22 15:57
 */
@Slf4j
@ChannelHandler.Sharable
public class ClientHandler extends ChannelDuplexHandler {
    // 使用 map 维护 id 和 Future 的映射关系，在多线程环境下需要使用线程安全的容器
    private final Map<String, DefaultFuture> futureMap = new ConcurrentHashMap<>();

//    @Override
//    public void channelActive(ChannelHandlerContext ctx) throws Exception {
//        System.out.println("通道活跃");
//        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,server", CharsetUtil.UTF_8));
//    }

    /**
     * description：发送请求时，建立键值对映射，方便后面使用wait-notify异步取值
     * @Date: 2021/3/22 14:41
     */
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
//        System.out.println("我是client端的write，开始写消息了");
        //判断，只对RequestRpc消息建立键值map映射
        if(msg instanceof RpcRequest){
            RpcRequest request = (RpcRequest) msg;
            futureMap.putIfAbsent(request.getRequestId(),new DefaultFuture());
        }
        super.write(ctx, msg, promise);
    }

    /**
     * description：异步接收到服务端响应，
     * @Date: 2021/3/22 14:41
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //判断，只对ResponseRpc消息建立键值map映射
        if(msg instanceof RpcResponse){
            RpcResponse response = (RpcResponse) msg;
            log.info("当前的参数为：response={},futureMap={}",response,futureMap);
            DefaultFuture defaultFuture = futureMap.get(response.getRequestId());
            defaultFuture.setResponse(response);
        }
        super.channelRead(ctx, msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause);
        super.exceptionCaught(ctx, cause);
    }

    public RpcResponse getRpcResponse(String requestId) {
        try {
            DefaultFuture future = futureMap.get(requestId);
            return future.getResponse(5000);
        } finally {
            //获取成功以后，从map中移除
            futureMap.remove(requestId);
        }
    }
}
